#include <stdio.h>
#include <errno.h>
#include <stdint.h>
#include <sys/eventfd.h>

#include "php.h"
#include "factory.h"
#include "server.h"
#include "silent.h"
#include "utils.h"

extern jmp_buf jumper;

#ifndef MAX
#define MAX(a, b)  (((a)>(b))?(a):(b))
#endif

#define SLT_MQNAME	"/silent_mqname"
#define SLT_WORKER_INIT_STARTUP(factory)	for (; i<factory->fork_num; i++) {	\
                factory->child_ids[i] = factory->spawn(factory);							\
            }

void *replysock_checkfd(void *data);

#define SLT_FIND_SLAB_BLOCK(len)	if (len <= 88) {	\
        ele = data_pool->ele_88_200;				\
        max_pos = 200;								\
    } else if (len <= 176) {						\
        ele = data_pool->ele_176_190;				\
        max_pos = 190;								\
    } else if (len <= 352) {						\
        ele = data_pool->ele_352_180;				\
        max_pos = 180;								\
    } else if (len <= 704) {						\
        ele = data_pool->ele_704_170;				\
        max_pos = 170;								\
    } else if (len <= 1408) {						\
        ele = data_pool->ele_1408_160;				\
        max_pos = 160;								\
    } else if (len <= 5632) {						\
        ele = data_pool->ele_5632_250;				\
        max_pos = 250;								\
    } else {										\
        ret = -1;									\
        goto alloc_err;								\
    }

typedef struct _thread_passdata {
	void *ptr;
	int mode;
} thread_passdata_t;

extern int slt_errno;
extern int silent_running;


static int slt_factory_dispatch(factory_t *factory, int fd, int schedid, char *data, int len) {
	slt_server_t *serv = factory->ptr;
	
	//write(fd, data, len);
	factory->notify(factory, fd, schedid, data, len);
	return 0;
}

static int slt_factory_worker(factory_t *factory, event_node *data_node) {
	slt_server_t *serv = factory->ptr;
	data_pool_t *data_pool = factory->data_pool;
	zval *php_zrcv = NULL;
	element_t *ele = NULL;
	off_t pos = 0;
	off_t max_pos;
	unsigned int max_size;
	int ret = 0;
	unsigned int len;

	php_zrcv = serv->call_rcv(data_node->from_id, data_node->data, data_node->len);
	len = Z_STRLEN_P(php_zrcv);

	if (len <= 0) {
		return 0;
	}


	data_pool->lock(&(data_pool->mutex));
	SLT_FIND_SLAB_BLOCK(len);
	
	for (; pos < max_pos; pos++) {
		if (ele->dirty == 1) {
			ele++;
			continue ;
		} else if (ele->ptr != NULL) {
			memcpy(ele->ptr, Z_STRVAL_P(php_zrcv), Z_STRLEN_P(php_zrcv));
			zval_ptr_dtor(&php_zrcv);
			ele->dirty = 1;
			ele->fd = data_node->fd;
        	ele->allocsize = len;
			ret = len;
			break;
		} else {
			ele->dirty = 1;
			ele->allocsize = len;
			ele->ptr = data_pool->mem;
			ele->fd = data_node->fd;
			data_pool->mem += len;
			memcpy(ele->ptr, Z_STRVAL_P(php_zrcv), Z_STRLEN_P(php_zrcv));
			ret = len;
			zval_ptr_dtor(&php_zrcv);
			break;
		}
	}
alloc_err:
	data_pool->unlock(&(data_pool->mutex));
	return ret;
}

int slt_factory_process(factory_t *factory, event_emit_node_t *emit_node) {
	char buf[8192];
    int n;
	slt_server_t *serv = factory->ptr;
	slt_thread_pool_handler_t *pool = serv->pool;
	reactor_t nginx_reactor;

    n = slt_utils_read(emit_node->fd, buf, 8192);
    if (n == -1) {

    } else if (n == 0) {
        //客户端已关闭
        close(emit_node->fd);
		nginx_reactor = pool[emit_node->from_id].reactor;
		nginx_reactor.del(&nginx_reactor, emit_node->fd);
	} else {
		factory->dispatch(factory, emit_node->fd, emit_node->from_id, buf, n);
		if (slt_errno == EAGAIN) {
			factory->process(factory, emit_node);
			return ;
		}
	}
	return 0;
}

static inline void slt_factory_notify(factory_t *factory, int fd, int schedid, char *data, int len) {
	mqd_t mqd;
	event_node data_node;

	data_node.from_id = schedid;
	data_node.fd = fd;
	memcpy(data_node.data, data, len);
	data_node.len = len;

	mqd = mq_open(factory->unicom.mqd_name, O_WRONLY);
	mq_send(mqd, (char *)&data_node, sizeof(event_node), 0);
	mq_close(mqd);
}

static pid_t slt_zombie_wait(int *stat_loc) {
	int retval;

	while (((retval = wait(stat_loc)) == -1) && (errno == EINTR)) {
		;
	}

	return retval;
}

static void slt_factory_backend_exit(factory_t *factory) {
	slt_server_t *serv = factory->ptr;

	serv->wait_jobnum = 0;
	longjmp(jumper, -1);
}

static pid_t slt_factory_spawn(factory_t *factory) {
	int ret;
	pid_t pid = fork();

	if (pid == 0) {
    	event_node data_node;
    	ssize_t mq_ret;
    	mqd_t mqd;
    	char *buff;
    	struct mq_attr mq_attr;

    	while (silent_running) {
			mqd = mq_open(factory->unicom.mqd_name, O_RDONLY);
			mq_getattr(mqd, &mq_attr);
			buff = emalloc(mq_attr.mq_msgsize);
			mq_ret = mq_receive(mqd, buff, mq_attr.mq_msgsize, NULL);
			memcpy(&data_node, buff, sizeof(event_node));
			efree(buff);
			mq_close(mqd);
			if ((mq_ret == -1 && errno == EINTR) || (!silent_running)) {
	    		continue ;
			}
			ret = factory->worker(factory, &data_node);
			if (ret == 0) {
				continue ;
			}
			factory->sync_evfd.notify(factory->sync_evfd.evfd, ret);
    	}
		slt_factory_backend_exit(factory);
	} else {
		return pid;
	}
}

static inline int slt_factory_mutex_lock(pthread_mutex_t *mutex) {
	return pthread_mutex_lock(mutex);
}

static inline int slt_factory_mutex_unlock(pthread_mutex_t *mutex) {
	return pthread_mutex_unlock(mutex);
}

static inline int slt_factory_mutex_init(pthread_mutex_t *mutex, pthread_mutexattr_t *attr) {
	pthread_mutexattr_init(attr);
	pthread_mutexattr_setpshared(attr, PTHREAD_PROCESS_SHARED);
	pthread_mutex_init(mutex, attr);
	pthread_mutexattr_destroy(attr);
}

static int slt_factory_evfd_wait(int evfd) {
	uint64_t wait_data;
	ssize_t ret;

	while (silent_running) {
		ret = read(evfd, &wait_data, sizeof(uint64_t));
		if (ret == -1 && errno == EINTR) {
			continue ;
		}
		break;
	}
	return wait_data;
}

inline int slt_factory_evfd_notify(int evfd, int len) {
	uint64_t u = 1;

	return write(evfd, &u, sizeof(uint64_t));
}

static void *slt_factory_waitworker_thread(void *args) {
        thread_passdata_t *passdata = args;
        factory_t *factory = passdata->ptr;
	int i, ret, len = passdata->mode;
	int les[] = {88, 176, 352, 704, 1408, 5632};
	slt_server_t *serv = factory->ptr;
        data_pool_t *data_pool = factory->data_pool;
        element_t *ele = NULL;
        off_t pos, max_pos;
	char buff[8192];

	efree(args);
reflush_wait:
	pos = 0;
        factory->sync_evfd.wait(factory->sync_evfd.evfd);
	if (silent_running == 0) goto force_exit;
	for (i=0; i<6; i++) {
		SLT_FIND_SLAB_BLOCK(les[i]);
		data_pool->lock(&(data_pool->mutex));
		for (pos = 0; pos < max_pos; pos++) {
			if (ele->dirty == 1) {
				bzero(buff, sizeof(buff));
				memcpy(buff, ele->ptr, ele->allocsize);
				ele->dirty = 0;
				factory->done(factory, buff, ele->allocsize, ele->fd);
			}
			ele++;
		}
		data_pool->unlock(&(data_pool->mutex));
	}
alloc_err:
	goto reflush_wait;
force_exit:
	pthread_exit((void *)NULL);
}

static inline int slt_factory_mutex_destroy(pthread_mutex_t *mutex) {
	return pthread_mutex_destroy(mutex);
}

static inline void slt_factory_syncevfd_init(factory_t *factory) {
	factory->sync_evfd.evfd = eventfd(0, EFD_SEMAPHORE);
        factory->sync_evfd.wait = slt_factory_evfd_wait;
        factory->sync_evfd.notify = slt_factory_evfd_notify;
}

void slt_factory_start(factory_t *factory) {
	pid_t pid;
	mqd_t mqd;
	pthread_t checkfdtid;
	void *mem = NULL;
	pthread_mutexattr_t attr;
	slt_server_t *serv = factory->ptr;
	data_pool_t **data_pool = &(factory->data_pool);
	data_pool_t *ref_pool = NULL;
	unsigned int i = 0;
	size_t element_t_size = sizeof(element_t);
	struct mq_attr mq_attr;

	zend_llist_init(&(factory->reply_sock_manage), sizeof(reply_sock_node_t), NULL, 0);

	pthread_create(&checkfdtid, NULL, replysock_checkfd, factory);
	pthread_detach(checkfdtid);
	slt_factory_syncevfd_init(factory);

	factory->fork_num = serv->job_num;
	factory->child_ids = ecalloc(factory->fork_num, sizeof(pid_t));
	mq_attr.mq_maxmsg = 10;
	mq_attr.mq_msgsize = (sizeof(int) * 3) + 8192;

	mqd = mq_open(SLT_MQNAME, O_RDWR|O_CREAT|O_EXCL, 0644, &mq_attr);
	mq_close(mqd);
	memcpy(factory->unicom.mqd_name, SLT_MQNAME, sizeof(SLT_MQNAME));

	mem = mmap(NULL, MAP_MEMSIZE, PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
	bzero(mem, MAP_MEMSIZE);
	*data_pool = mem;
	ref_pool = mem;
	ref_pool->init_mem = mem;
	
	ref_pool->init = slt_factory_mutex_init;
        ref_pool->lock = slt_factory_mutex_lock;
        ref_pool->unlock = slt_factory_mutex_unlock;
	ref_pool->destroy = slt_factory_mutex_destroy;

	ref_pool->ele_88_200 = mem + sizeof(data_pool_t);
	ref_pool->ele_176_190 = ref_pool->ele_88_200 + 200;
	ref_pool->ele_352_180 = ref_pool->ele_176_190 + 190;
	ref_pool->ele_704_170 = ref_pool->ele_352_180 + 180;
	ref_pool->ele_1408_160 = ref_pool->ele_704_170 + 170;
	ref_pool->ele_5632_250 = ref_pool->ele_1408_160 + 160;


	ref_pool->mem = ref_pool->ele_5632_250 + 250;
	ref_pool->init(&(ref_pool->mutex), &attr);

	pid = fork();
	switch (pid) {
		case -1:
			break;
		case 0: {
			pid_t child_pid;

			SLT_WORKER_INIT_STARTUP(factory);
watch_sub_worker:
			while (silent_running) {
				while ((child_pid = slt_zombie_wait(NULL)) != -1) {
					silent_running && factory->spawn(factory);
                    continue ;
                }
				goto watch_sub_worker;
			}
			ref_pool->destroy(&(ref_pool->mutex));
			slt_factory_backend_exit(factory);
		}
		default: {
			pthread_t tid;
			int i = 0;
			thread_passdata_t *passdata = NULL;
			unsigned int mode[] = {88, 176, 352, 704, 1408, 5632};
			factory->sub_main_id = pid;
			serv->wait_jobtids = ecalloc(serv->wait_jobnum, sizeof(pthread_t));
			for (; i < serv->wait_jobnum; i++) {
				passdata = emalloc(sizeof(thread_passdata_t));
				passdata->ptr = factory;
				passdata->mode = mode[i];
				pthread_create(&tid, NULL, slt_factory_waitworker_thread, passdata);
				serv->wait_jobtids[i] = tid;
			}
		}
	}
}

int checkfd_period(void *data) {
	reply_sock_node_t *node = NULL;
	time_t per_time;

	time(&per_time);
	node = data;
	if ((per_time - node->accept_time) > node->period) {
		shutdown(node->fd, SHUT_WR);	
		return 1;	
	}
	return 0;
}

void *replysock_checkfd(void *data) {
	factory_t *factory = data;

	while (silent_running) {
		zend_llist_apply_with_del(&(factory->reply_sock_manage), checkfd_period);
		usleep(60000);
	}
}

static void slt_factory_done(factory_t *factory, char *buff, int len,  int fd) {
	reply_sock_node_t *rsn = NULL;
	slt_server_t *serv = factory->ptr;
	time_t now;
	zval *zv = NULL;

	time(&now);
	rsn = emalloc(sizeof(reply_sock_node_t));
	rsn->accept_time = now;
	rsn->fd = fd;
	rsn->period = 1;

	zend_llist_add_element(&(factory->reply_sock_manage), rsn);
	efree(rsn);
	zv = serv->call_done(buff, len);
	write(fd, Z_STRVAL_P(zv), Z_STRLEN_P(zv));
	zval_ptr_dtor(&zv);
}

void slt_factory_create(slt_server_t *serv) {
	factory_t *factory = &(serv->factory);
	
	factory->ptr = serv;
	factory->dispatch = slt_factory_dispatch;
	factory->notify = slt_factory_notify;
	factory->process = slt_factory_process;
	factory->start = slt_factory_start;
	factory->worker = slt_factory_worker;
	factory->spawn = slt_factory_spawn;
	factory->done = slt_factory_done;
}
